package KafkaApps;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.LongSerializer;

import java.io.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ProducerFileWhole {
    private final static String TOPIC = "testTrans";
    private final static String FILEPATH = "trans_file/producer_file/kafka.jpg";

    private static Producer<Long, byte[]> createProducer(){
        return getProducer();
    }

    static Producer<Long, byte[]> getProducer() {
        Properties set = new Properties();
        set.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master:9092");
        set.put(ProducerConfig.CLIENT_ID_CONFIG, "ProducerFiles");
        set.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());
        set.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,ByteArraySerializer.class.getName());
        set.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,2048096);
        return new KafkaProducer<>(set);
    }

    static void runProducer(){
        final Producer<Long, byte[]> producer = createProducer();
        long time = System.currentTimeMillis();

        //读文件
        File file = new File(FILEPATH);
        FileInputStream fileInputStream = null;
        //FileOutputStream fileOutputStream = null;
        byte[] bytes = new byte[1024 * 10 * 10 * 10 * 2]; //2MB
        try {
            fileInputStream = new FileInputStream(file);
            //fileOutputStream = new FileOutputStream("src/main/producer_file/a.jpg");
            fileInputStream.read(bytes);
            //fileOutputStream.write(bytes);
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                fileInputStream.close();
                //fileOutputStream.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        //生产
        final ProducerRecord<Long, byte[]> producerRecord = new ProducerRecord<>(TOPIC, time, bytes);
        try {
            //RecordMetadata metadata = producer.send(producerRecord).get();
            RecordMetadata metadata = producer.send(producerRecord).get();

            System.out.printf("发送文件%s成功！meta(partition=%d, offset=%d) ", file.getName()
                    ,metadata.partition(),metadata.offset());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            producer.flush();
            producer.close();
        }
    }

    public static void main(String... args){
        runProducer();
    }
}
